package dataflowscheme.engine.core;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.log4j.Logger;
import org.eclipse.emf.common.util.EList;

import dataflowScheme.ConnectionType;
import dataflowScheme.DataPort;
import dataflowScheme.Port;
import dataflowScheme.PortSet;
import dataflowScheme.SynchroPort;

/**
 * The encapsulation class for processing entities Task and Storage.
 * An instance is used by several threads: Main thread uses
 * init() method, Event parser thread uses methods for receiving
 * signals and data (receiveData, receiveSignal) and its feedback
 * (receiveResponse), and Execution Event Parser uses receiveResult
 * method.
 * 
 * @author <a href="mailto:misiak7@gmail.com">Michal Antolik</a>
 *
 */
/**
 * @author <a href="mailto:misiak7@gmail.com">Michal Antolik</a>
 *
 */
public abstract class PE implements IPE {

	static protected Logger log = Logger.getLogger(PE.class);
	
	/** source class on which processing will be called */
	protected Class<?> source;
	
	/** an instance of 'source' used for execution */
	protected Object sourceInstance;
	
	/** it has value null if there was no execution or
	 * the previous execution ends successfully (all data and
	 * signals were sent and input data from portset used for
	 * execution were deleted). Otherwise, it has an integer
	 * number identifying an input portset which is currently
	 * executed.*/
	protected Integer lastInvokedPortSet;

	/** a method to be used for execution call on referenced class */
	protected Method runMethod;
	
	/** appropriate processing entity instance from input EMF scheme (ecore object)*/
	protected dataflowScheme.PE ePe;
	
	/** map of methods (of referenced class) for each data port */
	protected Map<String, Method> methods = new HashMap<String, Method>();

	/** map of connections for particular processing entity.
	 * It takes port name and return connected PE object and
	 * particular port name. */
	protected Map<String, ConnectionAnchor> oppossitePort = new HashMap<String, ConnectionAnchor>();
	
	/** map of data produced in execution phase. It takes
	 * port name and return produced data object */
	protected Map<String, Object> producedData = new HashMap<String, Object>();
	
	/** list of signals to be send. In execution phase a number of 
	 * output portset is selected, and this list contain port names
	 * which are intended to send signal to opposite port. */
	protected List<String> signals2send = new ArrayList<String>();
	
	/** queue of Event Parser to process events like sendSignal, sendData, ..*/
	protected final LinkedBlockingQueue<IEvent> eventQueue;
	
	/** queue of Execution Event Parser to process events like TaskExecution
	 * or StorageExecution event (to invoke execution of referenced class)*/
	protected final LinkedBlockingQueue<IExecutionEvent> executionQueue;
	
	protected static final Class<?>[] setParams = new Class<?>[]{java.lang.Object.class};
	protected static final Class<?>[] getParams = new Class<?>[0];
	
	public PE(dataflowScheme.PE ePe, LinkedBlockingQueue<IEvent> queue, LinkedBlockingQueue<IExecutionEvent> executionQueue){
		this.ePe = ePe;
		this.eventQueue = queue;
		this.executionQueue = executionQueue;
	}
	
	/**It creates Methods objects for main run method used for 
	 * execution and also it checks for an existence of all methods
	 * used to set or get methods from ports of referenced class.
	 * 
	 * @throws DFinitException when some method is not found
	 */
	public void init() throws DFinitException {
		String className = getClassName(ePe.getSource());
		Class<?>[] runMethodParams = new Class<?>[]{java.lang.Integer.class};
		
		try{
			source = Class.forName(className);
			runMethod = source.getMethod("run", runMethodParams);

			for(PortSet ps : ePe.getPortSet()){
				processPortSetMethods(ps);
			}
		}catch(ClassNotFoundException cnfe){
			throw new DFinitException("PE '" + ePe.getName() + "' init interrupted - no class found", cnfe);
		} 
		catch(NoSuchMethodException nsme){
			throw new DFinitException("PE '" + ePe.getName() + "' init interrupted - no 'run' method found", nsme);
		}
	}
	
	
	/**process referenced source name from path form to "dot" form, 
	 * e.g. from "/loadResource/model-src/dataflow/example/Worker2.java" to
	 * "dataflow.example.Worker2"
	 * 
	 * @param class name of referenced Java class saved in 'Source' property 
	 * @return
	 */
	protected String getClassName(String resourceSource){
		String[] parts = resourceSource.split("/");
		StringBuilder className = new StringBuilder();
		for(int i = 3; i < parts.length - 1; i++){
			className.append(parts[i]);
			className.append(".");
		}
		String fileName = parts[parts.length-1];
		className.append(fileName.split("[.]")[0]);
		
		log.debug(msg(className.toString()));
		
		return className.toString();
	}
	
	/**check for all method existence in referenced source for 
	 * all ports of particular portSet
	 * 
	 * @param ePortSet (EMF ecore PortSet object)
	 * @throws DFinitException if some methods are not found
	 */
	protected void processPortSetMethods(PortSet ePortSet) throws DFinitException{
		
		EList<Port> ports = ePortSet.getPort();
		
		InputPortSetStruct pss = null;
		//we do not need special portset structure for output ports
		if(ePortSet.getType() == ConnectionType.IN){
			pss = getInputPortSetStruct(ePortSet.getId());
		}
		
		for(Port p : ports){
			String portName = p.getName();
			
			if(pss != null){
				//register place where token will be saved in runtime phase
				pss.registerPort(portName);
			}
			
			if(p instanceof DataPort){
				//check if referenced method exists for DataPort
				processPortMethod(p);
			}
		}
	}
	
	/**check for method existence in referenced source for particular 
	 * data port. Signal ports are not developed in referenced classes.
	 * 
	 * @param p ePort (EMF ecore port object)
	 * @throws DFinitException if a method is not accessible
	 */
	private void processPortMethod(Port p) throws DFinitException{
		String portName = p.getName();
		
		if(methods.containsKey(portName)){
			throw new DFinitException("Port name '" + portName + "' is duplicated.");
		}

		boolean in = p.getType().equals(ConnectionType.IN);
		String methodPrefix = in?"set":"get";
		String methodName = methodPrefix + portName;
		
		try{
			Method m = source.getMethod(methodName, in?setParams:getParams);
			methods.put(portName, m);
		} catch (NoSuchMethodException nsme){
			throw new DFinitException("Failed while looking for method '" + methodName + "' of '" + ePe.getName() + "'", nsme);
		}
	}

	/**
	 * It is used by a thread from Execution event parser's 
	 * thread pool to send signal and data tokens. 'Send' means
	 * to create Synchro or Data Events. 
	 */
	protected void sendResult(){
		
		//create copy of lists of tokens to be send - it is necessary because
		//this method (sending data) and method which handles data receiving
		//are processed in two different threads - while this thread is sending 
		//events, the event parser can already process them
		Map<String, Object> producedDataCopy = new HashMap<String, Object>();
		producedDataCopy.putAll(producedData);
		
		List<String> signals2sendCopy = new ArrayList<String>();
		signals2sendCopy.addAll(signals2send);
		
		for(Map.Entry<String, Object> e : producedDataCopy.entrySet()){
			sendData(e.getValue(), e.getKey());
		}

		//send signals for stoppers as last
		List<String> ports4stoppers = new ArrayList<String>();
		for(String signalPort : signals2sendCopy){
			if(oppossitePort.get(signalPort).pe instanceof CStopper){
				ports4stoppers.add(signalPort);
			} else {
				sendSignal(signalPort);
			}	
		}

		for(String signalPort : ports4stoppers){
			sendSignal(signalPort);
		}
	}
	
	/**
	 * Create DataEvent to send data from a port to an opposite port.
	 * It is processed by both Event parser thread and one of execution
	 * event parser threads.
	 * 
	 * @param data A data token
	 * @param portName A port name, from which data should be send
	 */
	protected void sendData(Object data, String portName) {
		ConnectionAnchor in = oppossitePort.get(portName);
		ConnectionAnchor out = new ConnectionAnchor(this, portName);
		
		try{
			eventQueue.put(new DataEvent(out, data, in));
		} catch(InterruptedException ie){
			log.error(msg("Trying to send data event when event queue is already dead."), ie);
		}
	}
	
	/**
	 * Create SynchroEvent to send signal from a port to an opposite
	 * port. It is processed by both Event parser thread and one of execution
	 * 
	 * @param portName A port name, from which signal should be send
	 */
	protected void sendSignal(String portName){
		ConnectionAnchor in = oppossitePort.get(portName);
		ConnectionAnchor out = new ConnectionAnchor(this, portName);
		
		try{
			eventQueue.put(new SynchroEvent(out, in));
		} catch(InterruptedException ie){
			log.error(msg("Trying to send synchro event when event queue is already dead."), ie);
		}
	}

	/**
	 * Prepare a processing entity to be able to start new execution. 
	 * It removes tokens from all ports of input portset which was used
	 * for an execution.
	 */
	protected void cleanBeforeNewExecution(){
		log.debug(msg("cleaning for next execution"));
		if(lastInvokedPortSet != null){
			getInputPortSetStruct(lastInvokedPortSet).executionEnded();
			lastInvokedPortSet = null;
		}
	}
	
	/* (non-Javadoc)
	 * @see dataflowscheme.engine.core.IConnectableEntity#receiveSignal(String portName)
	 */
	public boolean receiveSignal(String portName) {
		InputPortSetStruct pss = inputPortsetNameMap.get(portName);
		if(!pss.hasPort(portName)){
			log.error(msg("Trying to set data to unknown port"));
			return false;
		}
		
		if(pss.hasSignal(portName)){
			log.debug(msg("Token already exist in port " + portName + ", therefore rejecting attempt"));
			pss.setFailedAttempt(portName);
			return false;
		} else {
			pss.addSignalToken(portName); //announce new token
		}
		
		if(pss.isReady4execution()){
			invokeNewExecution(pss.id);
		}
		
		return true;
	}
	
	/* (non-Javadoc)
	 * @see dataflowscheme.engine.core.IConnectableEntity#receiveData(Object data, String portName)
	 */
	public boolean receiveData(Object data, String portName) {
		InputPortSetStruct pss = inputPortsetNameMap.get(portName);
			
		if(!pss.hasPort(portName)){
			log.error(msg("Trying to set data to unknown port '" + portName + "'."));
			return false;
		}
		
		if(pss.hasData(portName)){
			log.debug(msg("Token already exist in port '" + portName + "', therefore rejecting attempt"));
			pss.setFailedAttempt(portName);
			return false;
		} else {
			pss.addDataToken(portName, data);
		}
		
		if(pss.isReady4execution()){
			invokeNewExecution(pss.id);
		}
		
		return true;
	}
	
	/**It is intended to queue portSet numbers which are ready 
	 * to be executed but certain execution is already running  */
	protected Queue<Integer> iterationQueue = new LinkedList<Integer>();
	
	/**
	 * It schedules a new demand for an execution or queue it to
	 * 'iterationQueue' if a certain portSet is already executing
	 * 
	 * @param portSet number of portSet to be executed 
	 */
	protected void invokeNewExecution(int portSet){
		if(lastInvokedPortSet != null){
			//plan iteration
			log.debug(msg("queueing portset " + portSet));
			iterationQueue.add(portSet);
		} else {
			log.debug(msg("scheduling portset " + portSet));
			lastInvokedPortSet = portSet;
			schedule(portSet);
		}
	}
	

	/* (non-Javadoc)
	 * @see dataflowscheme.engine.core.IConnectableEntity#receiveResponse(EventMsg event, String portName)
	 */
	public void receiveResponse(EventMsg event, String portName) {
		switch(event){
		case freePort:
			ConnectionAnchor ca = oppossitePort.get(portName);
			log.debug(msg("resending token '"+portName+"' to '"+ca.pe.getName()+"/"+ca.portName+"'."));
			if(producedData.containsKey(portName)){
				sendData(producedData.get(portName), portName);
			} else if(signals2send.contains(portName)){
				sendSignal(portName);
			}
			break;
		case delivered_ok:
			if(!signals2send.remove(portName)){
				producedData.remove(portName);
			}
			
			if(producedData.isEmpty() && signals2send.isEmpty()){
				log.debug(msg("All data send, preparing new excution"));
				cleanBeforeNewExecution();
				
				//is any portSet waiting for an execution?
				Integer psId = iterationQueue.poll();
				if(psId != null){
					invokeNewExecution(psId);
				}
			}
			break;
		case cleanPEmanually:
			cleanBeforeNewExecution();
			
			//is any portSet waiting for an execution?
			Integer psId = iterationQueue.poll();
			if(psId != null){
				invokeNewExecution(psId);
			}
			break;
		}
	}

	/* (non-Javadoc)
	 * @see dataflowscheme.engine.core.IConnectableEntity#addConnection(String portName, IConnectableEntity oppossiteEntity, String oppossitePortName)
	 */
	public void addConnection(String portName,
			IConnectableEntity oppossiteEntity, String oppossitePortName) {
		oppossitePort.put(portName, new ConnectionAnchor(oppossiteEntity, oppossitePortName));
		
	}

	/* (non-Javadoc)
	 * @see dataflowscheme.engine.core.IPE#receiveResult(Integer returnPortset)
	 */
	public void receiveResult(Integer returnPortset) {
		try{
			boolean nothing2send = parseResult(returnPortset);
			if(nothing2send){
				log.debug(msg("No data to be send, invoking execution manually"));
				eventQueue.add(new CleanEvent(this));
			} else {
				sendResult();
			}
		} catch(Exception e){
			log.error(msg("Error while parsing result"), e);
		}
	}
	
	/**
	 * This method will fetch data from an instance of referenced
	 * class for a particular portSet.
	 *  
	 * @param returnPortset output portSet number, data will be fetched from its data ports 
	 * @return True if there are not data or signal tokens to be send
	 */
	protected boolean parseResult(Integer returnPortset){
		log.debug(msg("Parsing result"));
		if(returnPortset != null){
			
			PortSet ps = findPortSetByID(ePe, returnPortset);
			if(ps == null){
				log.error(msg("Bad number of portset."));
				return true;
			}
		
			for (Port p : ps.getPort()){
				if(p instanceof SynchroPort){
					signals2send.add(p.getName());
					continue;
				}
				Method getMethod = methods.get(p.getName());
				try{
					Object value = getMethod.invoke(sourceInstance, new Object[0]);
					producedData.put(p.getName(), value);
				}catch(InvocationTargetException ite){
					log.error(msg("Invocation of port '" + p.getName() + "' failed."), ite);
				} catch (IllegalArgumentException e) {
					log.error(msg("Method '" + getMethod.getName() + "' can't have any arguments."),e);
				} catch (IllegalAccessException e) {
					log.error(msg("Method '" + getMethod.getName() + "' have to be public."),e);
				} 
			}
		}
		return producedData.isEmpty() && signals2send.isEmpty();
	}
	
	/**
	 * @return appropriate processing entity instance from input EMF scheme (ecore object)
	 */
	public dataflowScheme.PE getPE(){
		return ePe;
	}

	protected abstract void schedule(int portset);
	
	/**
	 * @return source class on which processing will be called 
	 */
	public Class<?> getSourceClass(){
		return source;
	}
	
	/**
	 * @return map of methods (of referenced class) for each data port 
	 */
	public Map<String, Method> getPortMethods(){
		return methods;
	}
	
	/**
	 * @return a method to be used for execution call on referenced class 
	 */
	public Method getRunMethod(){
		return runMethod;
	}
	
	/**
	 * 
	 * @param pe EMF processing entity instance
	 * @param id number of portSet
	 * @return EMF portSet instance with ID='id'
	 */
	static public PortSet findPortSetByID(dataflowScheme.PE pe, int id){
		for (PortSet ps : pe.getPortSet()){
			if(ps.getId() == id){
				return ps;
			}
		}
		return null;
	}
	
	/**
	 * @return an instance of 'source' used for execution
	 */
	public Object getSourceInstance(){
		return sourceInstance;
	}
	
	/**
	 * Set an instance of 'source' used for execution
	 * 
	 * @param sourceInstance Referenced Java object
	 */
	public void setSourceInstance(Object sourceInstance){
		this.sourceInstance = sourceInstance;
	}
	
	/** map of InputPortSetStruct instances for portSet id */
	private Map<Integer, InputPortSetStruct> inputPortsets = new HashMap<Integer, InputPortSetStruct>();
	
	/**
	 * Returns InputPortSetStruct class instance for a portSet id. 
	 * If no such instance exist, then a new one is created.
	 * 
	 * @param id a portSet id
	 * @return InputPortSetStruct for 'id', never null
	 */
	public InputPortSetStruct getInputPortSetStruct(Integer id){
		InputPortSetStruct ps = null;
		if(inputPortsets.containsKey(id)){
			ps =  inputPortsets.get(id);
		} else {
			ps = new InputPortSetStruct(id);
			inputPortsets.put(id, ps);
		}
		return ps; 
	}
	
	/**
	 * Class to store received tokens. It also handles receiving and 
	 * rejecting  tokens if a port is full. 
	 * 
	 * @author <a href="mailto:misiak7@gmail.com">Michal Antolik</a>
	 *
	 */
	public class InputPortSetStruct{
		/** id of portSet set in Ecore data-flow scheme  */
		public final int id;
		
		/** the number of all portSet's ports */
		private int totalPorts;
		
		/** the number of occupied portSet's ports */
		private int occupiedPorts;
		
		/** map of received values for port names */
		private Map<String, Object> receivedValues = new HashMap<String, Object>();
		
		/** list of received signals for port names */
		private Set<String> receivedSignals = new HashSet<String>();
		
		/** list of failed attempts to set token for port names */
		private Set<String> failedAttempt = new HashSet<String>();
		
		public InputPortSetStruct(int id) {
			this.id = id;
		}
		
		/**
		 * An initialization of a port to the InputPortSetStruct
		 *  
		 * @param name port name
		 */
		public void registerPort(String name){
			inputPortsetNameMap.put(name, this);
			totalPorts++;
		}
		
		public void addSignalToken(String portname){
			receivedSignals.add(portname);
			occupiedPorts++;
		}
		
		public void addDataToken(String portname, Object value){
			receivedValues.put(portname, value);
			occupiedPorts++;
		}
		
		public boolean hasSignal(String portname){
			return receivedSignals.contains(portname);
		}
		
		public boolean hasData(String portname){
			return receivedValues.containsKey(portname);
		}
		
		public boolean hasPort(String name){
			return inputPortsetNameMap.containsKey(name);
		}
		
		public boolean isReady4execution(){
			return totalPorts == occupiedPorts;
		}
		
		public void removeToken(){
			occupiedPorts--;
		}
		
		public boolean isEmpty(){
			return occupiedPorts == 0;
		}
		
		public void setFailedAttempt(String portname){
			failedAttempt.add(portname);
		}
		
		public boolean removeFailedAttempt(String portname){
			return failedAttempt.remove(portname);
		}

		/**
		 * Clean input portSet: remove all data and signals, send feedback
		 * to opposite ports if they were unsuccessful to set data or signal before
		 */
		void executionEnded(){
			receivedSignals.clear();
			receivedValues.clear();
			occupiedPorts = 0;
			for(String port : failedAttempt){
				//announce data release from ports
				ConnectionAnchor ca = oppossitePort.get(port);
				log.debug(msg("port '" + port + "' is free, sending echo to '" + ca.pe.getName() + "/" + ca.portName + "'"));
				ca.pe.receiveResponse(EventMsg.freePort, ca.portName);
			}
			failedAttempt.clear();
		}
		
		public Map<String, Object> getReceivedValues(){
			return receivedValues;
		}
	}
	
	/** map of port name <-> InputPortSetStruct instance */
	protected Map<String, InputPortSetStruct> inputPortsetNameMap = new HashMap<String, InputPortSetStruct>();
	
	private String msg(String msg){
		return ePe.getName() + " - " + msg; 
	}
	
	/* (non-Javadoc)
	 * @see dataflowscheme.engine.core.IConnectableEntity#getName()
	 */
	public String getName(){
		return ePe.getName();
	}
}
